package com.dtruth.dataflow;

import java.util.HashMap;
import java.util.Map;

import org.apache.thrift7.TException;
import org.json.simple.JSONValue;

import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.generated.Nimbus;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.utils.NimbusClient;

public class LocalSubmit {

	public static void main(String[]args){
		Config conf = new Config();
		conf.setNumWorkers(2);
		conf.setDebug(true);
		TopologyBuilder builder = new TopologyBuilder();
		// ...
		// topology 其他配置信息等

		// 读取本地 Storm 配置文件
//		Map stormConf = Utils.readStormConfig();
		Map<String,Object> stormConf = new HashMap<String,Object>();
		stormConf.put("nimbus.host", "hd124");
		stormConf.putAll(conf);

		Nimbus.Client client = NimbusClient.getConfiguredClient(stormConf).getClient();
		String inputJar = "E:\\workspace\\storm-demo\\target\\storm-demo-0.0.5-SNAPSHOT-shade.jar";
		String uploadedJarLocation = StormSubmitter.submitJar(stormConf, inputJar);
		String jsonConf = JSONValue.toJSONString(stormConf);
		
		try {
		NimbusClient nimbus = new NimbusClient(stormConf, "hd124", 6627);
		// 使用 StormSubmitter 提交 jar 包
		nimbus.getClient().submitTopology("remotetopology", uploadedJarLocation, jsonConf, builder.createTopology());
		} catch (AlreadyAliveException | InvalidTopologyException | TException e) {
			e.printStackTrace();
		}
	}
}
